package cn.itcast.streaming.source;

import cn.itcast.streaming.bean.VehicleInfoModel;
import cn.itcast.streaming.utils.DateUtil;
import cn.itcast.streaming.utils.JDBCUtil;
import com.mysql.jdbc.TimeUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.Statement;
import java.util.Date;
import java.util.HashMap;
import java.util.concurrent.TimeUnit;

//加载车辆基础信息表的数据
public class MysqlVehicleInfoFunction extends RichSourceFunction<HashMap<String, VehicleInfoModel>> {
    private static Logger logger = LoggerFactory.getLogger(MysqlVehicleInfoFunction.class);
    private ParameterTool parameterTool; //全局参数对象
    private Connection conn;
    private PreparedStatement statement;
    private Boolean isRunning = true;

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        parameterTool = (ParameterTool) getRuntimeContext().getExecutionConfig().getGlobalJobParameters();
        conn = JDBCUtil.getConnection();
        statement = conn.prepareStatement("select t12.vin,t12.series_name,t12.model_name,t12.series_code,t12.model_code,t12.nick_name,t3.sales_date product_date,t4.car_type\n" +
                " from (select t1.vin, t1.series_name, t2.show_name as model_name, t1.series_code,t2.model_code,t2.nick_name,t1.vehicle_id\n" +
                " from vehicle_networking.dcs_vehicles t1 left join vehicle_networking.t_car_type_code t2 on t1.model_code = t2.model_code) t12\n" +
                " left join  (select vehicle_id, max(sales_date) sales_date from vehicle_networking.dcs_sales group by vehicle_id) t3\n" +
                " on t12.vehicle_id = t3.vehicle_id\n" +
                " left join\n" +
                " (select tc.vin,'net_cat' car_type from vehicle_networking.t_net_car tc\n" +
                " union all select tt.vin,'taxi' car_type from vehicle_networking.t_taxi tt\n" +
                " union all select tp.vin,'private_car' car_type from vehicle_networking.t_private_car tp\n" +
                " union all select tm.vin,'model_car' car_type from vehicle_networking.t_model_car tm) t4\n" +
                " on t12.vin = t4.vin");
    }

    @Override
    public void close() throws Exception {
        super.close();
        if(statement!=null && !statement.isClosed()) statement.close();
        if(conn!=null && !conn.isClosed()) conn.close();

    }

    @Override
    public void run(SourceContext<HashMap<String, VehicleInfoModel>> sourceContext) throws Exception {
        while (isRunning){
            ResultSet rs = statement.executeQuery();
            HashMap<String, VehicleInfoModel> resultMap = new HashMap<>();
            while (rs.next()){
                String vin = rs.getString("vin");
                String seriesName = rs.getString("series_name");
                String seriesCode = rs.getString("series_code");
                String modelCode = rs.getString("model_code");
                String modelName = rs.getString("model_name");
                String nickName = rs.getString("nick_name");
                String productDate = rs.getString("product_date");
                String carType = rs.getString("car_type");
                String liveTime = null;
                if(StringUtils.isNotEmpty(productDate)){
                    //使用年限=当前日期-出售日期
                    liveTime = String.valueOf((new Date().getTime() - rs.getDate("product_date").getTime())/1000/3600/24/365);
                }
                if(vin==null) vin="未知";
                if(seriesName==null) seriesName="未知";
                if(seriesCode==null) seriesCode="未知";
                if(modelCode==null) modelCode="未知";
                if(modelName==null) modelName="未知";
                if(nickName==null) nickName="未知";
                if(productDate==null) productDate="未知";
                if(carType==null) carType="未知";
                if(liveTime==null) liveTime="未知";
                VehicleInfoModel vehicleInfoModel = new VehicleInfoModel(vin, seriesCode, seriesName, modelCode, modelName, nickName, productDate, carType, liveTime);

                resultMap.put(vin,vehicleInfoModel);
            }
            //判断是否存在车辆基础信息数据
            if(resultMap.isEmpty()){
                logger.info("从车两基础型信息表中获取的车辆数据为空");
            }else {
                logger.info("车辆基础信息为:"+resultMap);
                sourceContext.collect(resultMap);
            }

            //设置休眠时间
            TimeUnit.SECONDS.sleep(Long.parseLong(parameterTool.getRequired("vehinfo.millionseconds")));

        }
    }

    @Override
    public void cancel() {
        isRunning = false;
    }
}
